import java.io.File;
import java.io.FileOutputStream;
import java.io.IOException;
import java.io.InputStream;
import java.io.OutputStreamWriter;
import java.io.Writer;
import java.text.Normalizer;
import java.util.ArrayList;
import java.util.List;

import com.amazonaws.AmazonClientException;
import com.amazonaws.AmazonServiceException;
import com.amazonaws.auth.PropertiesCredentials;
import com.amazonaws.services.s3.AmazonS3;
import com.amazonaws.services.s3.AmazonS3Client;
import com.amazonaws.services.s3.model.GetObjectRequest;
import com.amazonaws.services.s3.model.ObjectMetadata;
import com.amazonaws.services.s3.model.PutObjectRequest;
import com.amazonaws.services.s3.model.S3Object;
import com.amazonaws.services.sqs.AmazonSQS;
import com.amazonaws.services.sqs.AmazonSQSClient;
import com.amazonaws.services.sqs.model.CreateQueueRequest;
import com.amazonaws.services.sqs.model.DeleteMessageRequest;
import com.amazonaws.services.sqs.model.Message;
import com.amazonaws.services.sqs.model.ReceiveMessageRequest;
import com.itextpdf.text.pdf.PdfReader;
import com.itextpdf.text.pdf.parser.PdfReaderContentParser;
import com.itextpdf.text.pdf.parser.SimpleTextExtractionStrategy;
import com.itextpdf.text.pdf.parser.TextExtractionStrategy;

public class ServerPdfExtractor {

	public static void main(String[] args) throws Exception {
		
		new ServerPdfExtractor();
	}
	
	private ServerPdfExtractor() throws Exception {
		
		AmazonSQS sqs = new AmazonSQSClient(new PropertiesCredentials(
				ServerPdfExtractor.class
						.getResourceAsStream("AwsCredentials.properties")));

		AmazonS3 s3 = new AmazonS3Client(
				new PropertiesCredentials(ServerPdfExtractor.class
						.getResourceAsStream("AwsCredentials.properties")));

		System.out.println("===========================================");
		System.out.println("Server PDF Extractor");
		System.out.println("===========================================\n");

		// Accessing queue
		System.out.println("Accessing SQS queue called PdfQueue.\n");
		CreateQueueRequest createQueueRequest = new CreateQueueRequest(
				"PdfQueue");
		String pdfQueueUrl = sqs.createQueue(createQueueRequest).getQueueUrl();

		// Receive messages
		System.out.println("Receiving messages from PdfQueue.\n");

		// Create bucket for paragraphs
		String bucketName = "pdfparagraphbucket";
		System.out.println("Creating bucket " + bucketName + ".");
		s3.createBucket(bucketName);

		while (true) {
			try {

				ReceiveMessageRequest receiveMessageRequest = new ReceiveMessageRequest(
						pdfQueueUrl);
				List<Message> messages = sqs.receiveMessage(
						receiveMessageRequest).getMessages();
				int i = 0;
				for (Message message : messages) {
					System.out.println("Processing message: "
							+ message.getBody());

					processPdfFile(s3, bucketName, message.getBody());

					// Delete a message
					System.out.println("Deleting the message.\n");
					String messageRecieptHandle = messages.get(i)
							.getReceiptHandle();
					sqs.deleteMessage(new DeleteMessageRequest(pdfQueueUrl,
							messageRecieptHandle));
					i++;
				}

			} catch (AmazonServiceException ase) {
				System.out
						.println("Caught an AmazonServiceException, which means your request made it "
								+ "to Amazon SQS, but was rejected with an error response for some reason.");
				System.out.println("Error Message:    " + ase.getMessage());
				System.out.println("HTTP Status Code: " + ase.getStatusCode());
				System.out.println("AWS Error Code:   " + ase.getErrorCode());
				System.out.println("Error Type:       " + ase.getErrorType());
				System.out.println("Request ID:       " + ase.getRequestId());
			} catch (AmazonClientException ace) {
				System.out
						.println("Caught an AmazonClientException, which means the client encountered "
								+ "a serious internal problem while trying to communicate with SQS, such as not "
								+ "being able to access the network.");
				System.out.println("Error Message: " + ace.getMessage());
			}
		}
	}

	private File createFile(String paragraph) throws IOException {
		File file = File.createTempFile("pdf-paragraph", ".txt");
		file.deleteOnExit();

		Writer writer = new OutputStreamWriter(new FileOutputStream(file));
		writer.write(paragraph);
		writer.close();

		return file;
	}
	
	private void processPdfFile(AmazonS3 s3, String bucketDestinationName, String pdfLocation) throws IOException 
	{
		String []split=pdfLocation.split(";");
		String bucketSourceName = split[0];
		String pdfName = split[1];
		
		/* to do */
		if(pdfName.equals("1_paper_3.pdf"))
			return;

		/* Download pdf file */
		AmazonS3 s3pdf = new AmazonS3Client(
				new PropertiesCredentials(ServerPdfExtractor.class
						.getResourceAsStream("AwsCredentials.properties")));
		S3Object object = s3pdf.getObject(new GetObjectRequest(bucketSourceName,pdfName));
		
		/* Pdf Method */
		List<ParagraphEntry> entries = extract(object.getObjectContent());
		
		int i=0;
		
		for(ParagraphEntry entry : entries)
		{
			System.out.println("Uploading a new paragraph to S3 from a file\n");
			ObjectMetadata objectMetadata = new ObjectMetadata();
			objectMetadata.addUserMetadata("pdfName", pdfName);
			objectMetadata.addUserMetadata("page", Integer.toString(entry.getPageNumber()));
			
			/* data cleansing */
			String text = entry.getText();
			text = Normalizer.normalize(text, Normalizer.Form.NFD).replaceAll("[^\\p{ASCII}]","");
			text = text.replaceAll("[:;,.]","").toLowerCase();
			
			s3.putObject(new PutObjectRequest(bucketDestinationName, pdfName
					+ i, createFile(pdfName + ";" + i + ";" + text)));
			i++;
		}
	}
	
	public static List<ParagraphEntry> extract(InputStream inputStream) throws IOException
	{
		List<ParagraphEntry> paragraphs = new ArrayList<ParagraphEntry>();

		PdfReader reader = new PdfReader(inputStream);
		PdfReaderContentParser parser = new PdfReaderContentParser(reader);
		TextExtractionStrategy strategy;
		
		for (int i = 1; i <= reader.getNumberOfPages(); i++)
		{
			strategy = parser.processContent(i, new SimpleTextExtractionStrategy());
			String[] sentences = strategy.getResultantText().split("\\.");
			for (int j = 0; j < sentences.length; j += 10) {
				String paragraph = "";
				for (int k = j; k < j + 10 && k < sentences.length; k++)
					paragraph += (sentences[k] + ".");
				paragraphs.add(new ParagraphEntry(i, paragraph));
//				System.out.println(paragraph);
//				System.out.println("==============================" + i + "============================== " + j);
			}
		}
		
		inputStream.close();
		reader.close();
		
		return paragraphs;
	}

}
